package com.nanohadoop.yarn;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * 节点管理器，负责管理单个节点上的资源和容器
 */
public class NodeManager implements Serializable {
    private static final long serialVersionUID = 1L;
    private static final Logger log = LoggerFactory.getLogger(NodeManager.class);

    private String nodeId;
    private Resource totalResource;
    private Resource availableResource;
    private List<Container> containers;
    private volatile boolean running;
    private ExecutorService executorService;

    public NodeManager(String nodeId, int memoryMB, int cpuCores) {
        this.nodeId = nodeId;
        this.totalResource = new Resource(memoryMB, cpuCores);
        this.availableResource = new Resource(memoryMB, cpuCores);
        this.containers = new ArrayList<>();
        this.running = false;
        this.executorService = Executors.newCachedThreadPool();
    }

    public NodeManager(String nodeId, Resource resource) {
        this.nodeId = nodeId;
        this.totalResource = resource;
        this.availableResource = resource;
        this.containers = new ArrayList<>();
        this.running = false;
        this.executorService = Executors.newCachedThreadPool();
    }

    /**
     * 启动节点管理器
     */
    public void start() {
        if (running) {
            return;
        }
        running = true;
        log.debug("NodeManager started on node: " + nodeId);
    }

    /**
     * 停止节点管理器
     */
    public void stop() {
        if (!running) {
            log.debug("NodeManager on node: " + nodeId + " is already stopped");
            return;
        }
        log.debug("Initiating stop for NodeManager on node: " + nodeId);
        // 首先检查是否有正在运行的容器
        if (!containers.isEmpty()) {
            log.debug("NodeManager on node: " + nodeId + " has " + containers.size() + " running containers, waiting for them to complete...");
            try {
                // 等待容器完成，最多等待10秒
                TimeUnit.SECONDS.sleep(10);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
        running = false;
        executorService.shutdown();
        try {
            if (!executorService.awaitTermination(5, TimeUnit.SECONDS)) {
                executorService.shutdownNow();
            }
        } catch (InterruptedException e) {
            executorService.shutdownNow();
        }
        log.debug("NodeManager stopped on node: " + nodeId);
    }

    /**
     * 检查节点是否有足够资源分配容器
     */
    public boolean hasEnoughResource(Resource required) {
        return availableResource.getMemoryMB() >= required.getMemoryMB() &&
               availableResource.getCpuCores() >= required.getCpuCores();
    }

    /**
     * 分配容器
     */
    public synchronized boolean allocateContainer(Container container) {
        Resource required = container.getResource();
        log.debug("NodeManager " + nodeId + ": attempting to allocate container " + container.getContainerId());
        log.debug("NodeManager " + nodeId + ": required resources - Memory: " + required.getMemoryMB() + "MB, CPU: " + required.getCpuCores() + " cores");
        log.debug("NodeManager " + nodeId + ": available resources - Memory: " + availableResource.getMemoryMB() + "MB, CPU: " + availableResource.getCpuCores() + " cores");
        log.debug("NodeManager " + nodeId + ": running status: " + running);
        
        if (!hasEnoughResource(required)) {
            log.debug("NodeManager " + nodeId + ": not enough resources for container " + container.getContainerId());
            return false;
        }
        
        if (!running) {
            log.debug("NodeManager " + nodeId + ": not running, cannot allocate container " + container.getContainerId());
            // 添加额外日志，追踪谁调用了stop()
            log.debug("NodeManager " + nodeId + ": stack trace for debugging:");
            Thread.dumpStack();
            return false;
        }
        
        if (executorService.isShutdown() || executorService.isTerminated()) {
            log.debug("NodeManager " + nodeId + ": executor service is shutdown or terminated, cannot allocate container " + container.getContainerId());
            return false;
        }
        
        // 减少可用资源
        availableResource.setMemoryMB(availableResource.getMemoryMB() - required.getMemoryMB());
        availableResource.setCpuCores(availableResource.getCpuCores() - required.getCpuCores());
        // 添加容器
        containers.add(container);
        log.debug("NodeManager " + nodeId + ": container " + container.getContainerId() + " added to containers list");
        // 启动容器执行
        startContainer(container);
        return true;
    }

    /**
     * 启动容器执行
     */
    private void startContainer(Container container) {
        if (!running) {
            return;
        }
        executorService.submit(() -> {
            try {
                log.debug("Container " + container.getContainerId() + " started on node " + nodeId);
                // 模拟容器执行时间
                TimeUnit.SECONDS.sleep(2);
                log.debug("Container " + container.getContainerId() + " completed on node " + nodeId);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                log.debug("Container " + container.getContainerId() + " interrupted on node " + nodeId);
            } finally {
                // 执行完成后释放容器
                releaseContainer(container.getContainerId());
            }
        });
    }

    /**
     * 释放容器资源
     */
    public void releaseContainer(String containerId) {
        for (int i = 0; i < containers.size(); i++) {
            Container container = containers.get(i);
            if (container.getContainerId().equals(containerId)) {
                // 增加可用资源
                Resource resource = container.getResource();
                availableResource.setMemoryMB(availableResource.getMemoryMB() + resource.getMemoryMB());
                availableResource.setCpuCores(availableResource.getCpuCores() + resource.getCpuCores());
                // 移除容器
                containers.remove(i);
                return;
            }
        }
    }

    /**
     * 检查节点管理器是否正在运行
     */
    public boolean isRunning() {
        return running;
    }

    // Getters and Setters
    public String getNodeId() { return nodeId; }
    public void setNodeId(String nodeId) { this.nodeId = nodeId; }
    public Resource getTotalResource() { return totalResource; }
    public void setTotalResource(Resource totalResource) { this.totalResource = totalResource; }
    public Resource getAvailableResource() { return availableResource; }
    public void setAvailableResource(Resource availableResource) { this.availableResource = availableResource; }
    public List<Container> getContainers() { return containers; }

    @Override
    public String toString() {
        return "NodeManager{" +
                "nodeId='" + nodeId + '\'' +
                ", totalResource=" + totalResource +
                ", availableResource=" + availableResource +
                ", containers=" + containers.size() +
                '}';
    }
}